package com.atguigu.gmall.realtime.test;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

/**
 * @author Felix
 * @date 2024/7/24
 * 该案例演示了FlinkCDC的用法
 */
public class Flink01_FlinkCDC {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);

        env.enableCheckpointing(3000);
        //TODO 2.创建MySqlSource
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall0221_config") // set captured database
                .tableList("gmall0221_config.t_user") // set captured table
                .username("root")
                .password("123456")
                //.startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();
        //TODO 3.读取数据 封装为流
        DataStreamSource<String> mysqlStrDS = env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
        //TODO 4.打印输出
        //"op":"r": {"before":null,"after":{"id":1,"name":"zs","age":20},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall0221_config","sequence":null,"table":"t_user","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1721789500667,"transaction":null}
        //"op":"c": {"before":null,"after":{"id":3,"name":"ww","age":40},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1721789713000,"snapshot":"false","db":"gmall0221_config","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":1026,"row":0,"thread":8,"query":null},"op":"c","ts_ms":1721789713613,"transaction":null}
        //"op":"u": {"before":{"id":3,"name":"ww","age":40},"after":{"id":3,"name":"ww","age":18},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1721789765000,"snapshot":"false","db":"gmall0221_config","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":1351,"row":0,"thread":8,"query":null},"op":"u","ts_ms":1721789765712,"transaction":null}
        //"op":"d": {"before":{"id":3,"name":"ww","age":18},"after":null,"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1721789805000,"snapshot":"false","db":"gmall0221_config","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":1681,"row":0,"thread":8,"query":null},"op":"d","ts_ms":1721789805484,"transaction":null}

        mysqlStrDS.print();
        //TODO 5.提交
        env.execute("Print MySQL Snapshot + Binlog");
    }

}
